package beautiful.butterfly.drds.binlog.zookeeper;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.utils.EnsurePath;
import org.apache.zookeeper.CreateMode;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class ZooKeepers
{

    // 需要记住PathChildrenCache
    // 一个唯一的路径，肯定对应着一个唯一的cache
    private static ConcurrentHashMap<String, PathChildrenCache> pathToPathChildrenCacheMap = new ConcurrentHashMap<String, PathChildrenCache>();

    public static boolean registerPathChildrenCacheListener(String path, PathChildrenCacheListener pathChildrenCacheListener)
    {
        boolean result = false;
        try
        {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework(), path, true);
            pathChildrenCache.start(StartMode.POST_INITIALIZED_EVENT);
            pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
            result = true;
            // 保留
            pathToPathChildrenCacheMap.put(path, pathChildrenCache);
        } catch (Exception e)
        {
            log.info(e.toString());
        }
        return result;
    }

    public static boolean unregisterPathChildrenCacheListener(String path, PathChildrenCacheListener pathChildrenCacheListener)
    {
        boolean result = false;
        try
        {
            PathChildrenCache pathChildrenCache = pathToPathChildrenCacheMap.remove(path);
            if (null != pathChildrenCache)
            {
                pathChildrenCache.getListenable().removeListener(pathChildrenCacheListener);
            }
            result = true;
        } catch (Exception e)
        {
            log.info(e.toString());
        }
        return result;
    }

    // get children
    public static List<String> getChildrenList(String path)
    {
        try
        {
            return CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework().getChildren().forPath(path);
        } catch (Exception e)
        {
            log.info(e.toString());
        }
        return null;
    }

    // ensure class
    public static void ensurePersistentPathWithNoValue(String path)
    {
        try
        {
            EnsurePath ensurePath = new EnsurePath(path);
            ensurePath.ensure(CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework().getZookeeperClient());
            log.info("ensure zk path: " + path + " succeed");
            // LoggerUtils.debug(logger,
            // "--------------------------------------");
        } catch (Exception e)
        {

        }
    }

    // 是否存在
    public static boolean exist(String path)
    {
        CuratorFramework curatorFramework = CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework();
        boolean exist = false;
        try
        {
            if (null != curatorFramework.checkExists().forPath(path))
            {
                exist = true;
            }
        } catch (Exception e)
        {
            log.info(e.toString());
        }
        return exist;
    }

    // 增加永久节点
    public static void createPersistent(String path, String value)
    {// 永久性的
        CuratorFramework curatorFramework = CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework();
        try
        {
            curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath(path, value.getBytes());
            // 创建永久路径
            log.info("path not exist,create persistent path: " + path + " with value" + value + " succeed");
            // LoggerUtils.debug(logger,
            // "--------------------------------------");
        } catch (Exception e)
        {
            log.info(e.toString());
        }
    }

    public static void upsertEphemeral(String path, String value)
    {
        if (ZooKeepers.exist(path))
        {
            ZooKeepers.update(path, value);
        } else
        {
            ZooKeepers.createEphemeral(path, value);
        }
    }

    // 增加临时节点
    public static void createEphemeral(String path, String value)
    {// 临时性的
        CuratorFramework curatorFramework = CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework();
        try
        {
            curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(path, value.getBytes());
            // 创建永久路径
            log.info("path not exist,create temp path: " + path + " with value" + value + " succeed");
            // LoggerUtils.debug(logger,
            // "--------------------------------------");
        } catch (Exception e)
        {
            log.info(e.toString());
        }
    }

    // 删
    public static void deletePath(String path)
    {
        try
        {
            CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework().delete().guaranteed().forPath(path);
            log.info("succeed to delete path:" + path);
        } catch (Exception e)
        {
            // LoggerUtils.debug(logger, e.toString());
        }
    }

    // 改
    public static void update(String path, String value)
    {
        long begin = System.currentTimeMillis();
        try
        {
            CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework().setData().forPath(path, value.getBytes());
        } catch (Exception e)
        {
            // LoggerUtils.error(logger, e.toString());
        }
        long end = System.currentTimeMillis();
        log.info("update zk data cost:" + (end - begin) + " ms " + path + ":" + value);
    }

    // 跟上面的代码完全一样，只不过为了区分是用在并行加速中的函数，没其它意思
    public static void parallelUpdate(String path, String value)
    {
        try
        {
            CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework().setData().forPath(path, value.getBytes());
        } catch (Exception e)
        {
            log.info(e.toString());
        }

    }

    // 查
    public static String getData(String path)
    {
        try
        {
            byte[] data = CuratorFrameworkClient.getCuratorFrameworkClient().getCuratorFramework().getData().forPath(path);
            return new String(data);
        } catch (Exception e)
        {
            log.info(e.toString());
        }
        return null;
    }

}
